package com.hl233.framework.utils;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Instant;
import java.util.*;
import java.util.concurrent.ExecutionException;
/**
 * 适用于在应用程序代码中动态调整偏移量。
 * @author 张振强15065206763
 * @create 2024-09-04
 */
public class OffsetResetExample {

//    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        Properties adminProps = new Properties();
//        adminProps.put("bootstrap.servers", "127.0.0.1:9092");
//
//        try (AdminClient adminClient = AdminClient.create(adminProps)) {
//            String groupId = "defaul_group";
//            Instant targetTimestamp = Instant.parse("2024-09-03T12:00:00Z"); // 替换为目标时间
//
//            List<TopicPartition> partitions = new ArrayList<>();
//            // 添加需要重置偏移量的Topic和分区，例如：
//            partitions.add(new TopicPartition("my-topic", 0));
//
//            Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
//            for (TopicPartition partition : partitions) {
////                offsetSpecs.put(partition, OffsetSpec.forTimestamp(targetTimestamp));
//            }
//
////            adminClient.resetOffsets(groupId, offsetSpecs).all().get();
//            System.out.println("Offsets have been reset.");
//        }
//    }
}
